package com.zlll.winner.RabbitMqListener;

import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import com.zlll.winner.business.model.activity.ActivityManagement;
import com.zlll.winner.business.model.mall.MallOrder;
import com.zlll.winner.business.service.activity.IActivityManagementService;
import com.zlll.winner.business.service.mall.IMallOrderService;
import com.zlll.winner.business.service.media.IMediaListenUserLogService;
import com.zlll.winner.config.RabbitDelayConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.support.atomic.RedisAtomicLong;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * 延时监听器
 */
@Component
@Slf4j
public class RabbitMqDelayReceiver {

    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private IActivityManagementService managementService;
    @Autowired
    private IMallOrderService orderService;
    @Autowired
    private IMediaListenUserLogService listenUserLogService;


    @RabbitListener(queues = {RabbitDelayConfig.ACTIVE_DELAY_QUEUE})
    @Transactional(rollbackFor = {RuntimeException.class, Exception.class})
    public void activeDelayQueue(ActivityManagement management, Message message, Channel channel) throws Exception {
        Long id = management.getId();
        try {

            System.out.println("延迟r----------->" + System.currentTimeMillis());
            managementService.updateStatus(management.getId());
            //通知RabbitMQ消息已被接收,可以ACK(从队列中删除)了
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception ex) {
            try {
                RedisAtomicLong entityIdCounter = new RedisAtomicLong("activeRetry-" + id, redisTemplate.getConnectionFactory());
                long retry = entityIdCounter.getAndIncrement();
                if (retry > 9) {
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                    entityIdCounter.expire(0, TimeUnit.MILLISECONDS);
                    Integer status = management.getStatus();
                    id = management.getId();
                    log.error("【activeDelayQueue收到消息失败】-【活动延时内容】- [活动id={},当前status={}]", id, status);
                } else {
                    //重试10次
                    log.error("消费处理信息失败,尝试把消息重发给所有消费者,当前重试次数{}", retry);
                    channel.basicRecover(true);
                }
            } catch (IOException o) {
                log.error("{}", o.getMessage());
            } finally {
                //抛出异常，数据库回滚
                throw new Exception(ex.getMessage());
            }
        }
    }

    @RabbitListener(queues = {RabbitDelayConfig.MALL_DELAY_QUEUE})
    @Transactional(rollbackFor = {RuntimeException.class, Exception.class})
    public void mallDelayQueue(MallOrder order, Message message, Channel channel) throws Exception {
        Long id = order.getId();
        try {

            System.out.println("延迟mallDelayQueue----------->" + System.currentTimeMillis());
            orderService.updateOrderStatus(order.getId());
            //通知RabbitMQ消息已被接收,可以ACK(从队列中删除)了
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (Exception ex) {
            try {
                RedisAtomicLong entityIdCounter = new RedisAtomicLong("activeRetry-" + id, redisTemplate.getConnectionFactory());
                long retry = entityIdCounter.getAndIncrement();
                if (retry > 9) {
                    channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
                    entityIdCounter.expire(0, TimeUnit.MILLISECONDS);
                    log.error("【mallDelayQueue收到消息失败】-【活动延时内容】- [活动id={},当前status={}]", id, order.getStatus());

                } else {
                    //重试10次
                    log.error("消费处理信息失败,尝试把消息重发给所有消费者,当前重试次数{}", retry);
                    channel.basicRecover(true);
                }
            } catch (IOException o) {
                log.error("{}", o.getMessage());
            } finally {
                //抛出异常，数据库回滚
                throw new Exception(ex.getMessage());
            }
        }
    }


    /**
     * 消费直播日志记录任务队列
     * @param params
     * @param message
     * @param channel
     * @throws Exception
     */
    @RabbitListener(queues = {RabbitDelayConfig.LIVE_LISTEN_LOG_QUEUE})
    @Transactional(rollbackFor = {RuntimeException.class, Exception.class})
    public void liveListenLogQueue(JSONObject params, Message message, Channel channel) throws Exception {
        try {
            Object o = listenUserLogService.saveListenInLog(params);
            log.error("消费直播日志记录返回值===》{}",JSONObject.toJSONString(o));
        } catch (Exception ex) {
            log.error("消费直播日志记录异常==》{}",ex.getMessage());
        }finally {
            //通知RabbitMQ消息已被接收,可以ACK(从队列中删除)了
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        }
    }
}
